package com.boot.study.sink

import com.boot.study.api.SensorReading
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}

object RedisSinkTest {
  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //    // 读取文件
    val inputSteam: DataStream[String] = env.readTextFile("D:\\WorkSpace\\idea\\Flink\\src\\main\\resources\\sensor.txt")
    val dataStream: DataStream[SensorReading] = inputSteam.map(data => {
      val arr: Array[String] = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })

    dataStream.print("redis sink")

    // 定义一个FlinkJedisPoolConfig
    val conf: FlinkJedisPoolConfig = new FlinkJedisPoolConfig.Builder()
      .setHost("127.0.0.1")
      .setPort(6379)
      .build()

    dataStream.addSink(new RedisSink[SensorReading](conf, new MyRedisMapper))
    // 执行
    env.execute("redis sink test")
  }
}

// 定义一个redisMapper
class MyRedisMapper extends RedisMapper[SensorReading] {
  // 定义数据保存的命令 HashSet 表名 key value
  override def getCommandDescription: RedisCommandDescription = new RedisCommandDescription(RedisCommand.HSET, "sensor_temp")

  // 将id值指定为value
  override def getKeyFromData(data: SensorReading): String = data.id

  // 将温度值指定为value
  override def getValueFromData(data: SensorReading): String = data.temperature.toString
}
